package com.fiona.remoting.netty.client;

import java.net.InetSocketAddress;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.fiona.common.exception.RemotingException;
import com.fiona.remoting.api.AbstractClient;
import com.fiona.remoting.api.Constants;
import com.fiona.remoting.netty.NettyChannel;
import com.fiona.remoting.netty.NettyEventLoopFactory;
import com.fiona.remoting.netty.codec.MessageDecoder;
import com.fiona.remoting.netty.codec.MessageEncoder;

import cn.hutool.core.net.NetUtil;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;

public class NettyClient extends AbstractClient {
	private Logger logger = LoggerFactory.getLogger(NettyClient.class);
	private Bootstrap bootstrap;
	
	private static final EventLoopGroup NIO_EVENT_LOOP_GROUP = NettyEventLoopFactory.eventLoopGroup(Constants.DEFAULT_IO_THREADS, "NettyClientWorker");

	private volatile Channel channel;

	public NettyClient(InetSocketAddress address) throws RemotingException {
		super(address);
	}

	@Override
	protected void doOpen() throws Throwable {
		NettyClientHandler nettyClientHandler = new NettyClientHandler(this);

		bootstrap = new Bootstrap();
		bootstrap.group(NIO_EVENT_LOOP_GROUP).channel(NioSocketChannel.class).handler(new LoggingHandler(LogLevel.INFO))
				.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000).handler(new ChannelInitializer<SocketChannel>() {
					@Override
					protected void initChannel(SocketChannel ch) {
						ChannelPipeline p = ch.pipeline();
						p.addLast(new IdleStateHandler(0, 5, 0, TimeUnit.SECONDS));
						p.addLast(new MessageEncoder());
						p.addLast(new MessageDecoder());
						p.addLast(nettyClientHandler);
					}
				});

	}

	@Override
	public void received() {
		// TODO Auto-generated method stub

	}

	@Override
	protected void doClose() throws Throwable {
	}

	@Override
	protected void doConnect(InetSocketAddress inetSocketAddress) throws Throwable {
		long start = System.currentTimeMillis();
		ChannelFuture future = bootstrap.connect(getConnectAddress());
		try {
			boolean ret = future.awaitUninterruptibly(getConnectTimeout(), TimeUnit.MILLISECONDS);

			if (ret && future.isSuccess()) {
				Channel newChannel = future.channel();
				try {
					// Close old channel
					// copy reference
					Channel oldChannel = NettyClient.this.channel;
					if (oldChannel != null) {
						try {
							if (logger.isInfoEnabled()) {
								logger.info("Close old netty channel " + oldChannel + " on create new netty channel "
										+ newChannel);
							}
							oldChannel.close();
						} finally {
							NettyChannel.removeChannelIfDisconnected(oldChannel);
						}
					}
				} finally {
					if (NettyClient.this.isClosed()) {
						try {
							if (logger.isInfoEnabled()) {
								logger.info("Close new netty channel " + newChannel + ", because the client closed.");
							}
							newChannel.close();
						} finally {
							NettyClient.this.channel = null;
							NettyChannel.removeChannelIfDisconnected(newChannel);
						}
					} else {
						NettyClient.this.channel = newChannel;
					}
				}
			} else if (future.cause() != null) {
				throw new RemotingException("client failed to connect to server " + getConnectAddress()
						+ ", error message is:" + future.cause().getMessage(), future.cause());
			} else {
				throw new RemotingException("client failed to connect to server " + getConnectAddress()
						+ " client-side timeout " + getConnectTimeout() + "ms (elapsed: "
						+ (System.currentTimeMillis() - start) + "ms) from netty client " + NetUtil.getLocalhostStr());
			}
		} finally {
			// just add new valid channel to NettyChannel's cache
			if (!isConnected()) {
				// future.cancel(true);
			}
		}
	}

	@Override
	protected void doDisConnect() throws Throwable {
		try {
			NettyChannel.removeChannelIfDisconnected(channel);
		} catch (Throwable t) {
			logger.warn(t.getMessage());
		}
	}

	@Override
	protected com.fiona.remoting.api.Channel getChannel() {
		Channel c = channel;
		if (c == null) {
			return null;
		}
		return NettyChannel.getOrAddChannel(channel);
	}

	Channel getNettyChannel() {
		return channel;
	}

	int getConnectTimeout() {
		return 10*1000;
	}
}
